package realtime.dwd;

import org.apache.flink.table.api.Table
import BaseConfig._

/**
 * DWD层：数据仓库明细层，负责数据清洗、单位统一、异常值过滤
 * 工单编号：大数据-用户画像-11-达摩盘基础特征
 */
object DwdLayer {
        /**
         * 清洗用户行为数据：过滤刷单行为、筛选近30天行为
         * 刷单行为定义：短时间内高频浏览/加购非目标品类{insert\_element\_5\_}
         */
        def cleanUserBehavior(tEnv: StreamTableEnvironment): Table = {
        tEnv.sqlQuery(
        s"""
         |WITH user_behavior_stats AS (
         |  -- 统计用户1小时内行为频次，识别刷单（如1小时内>50次非目标品类行为）
         |  SELECT
         |    user_id,
         |    COUNT(*) AS hourly_behavior_cnt,
         |    TUMBLE_START(action_ts, INTERVAL '1' HOUR) AS hour_window
         |  FROM dwd_user_behavior_raw
         |  WHERE category_id NOT IN (SELECT category_id FROM dim_target_category) -- 非目标品类
         |  GROUP BY user_id, TUMBLE(action_ts, INTERVAL '1' HOUR)
         |),
         |cheating_users AS (
         |  SELECT DISTINCT user_id
         |  FROM user_behavior_stats
         |  WHERE hourly_behavior_cnt > 50 -- 刷单行为阈值
         |)
         |-- 输出非刷单用户的近30天行为明细
         |SELECT
         |  ub.user_id,
         |  ub.behavior_type,
         |  ub.category_id,
         |  ub.brand_id,
         |  ub.price,
         |  ub.search_keyword,
         |  ub.action_ts,
         |  ub.device_type
         |FROM dwd_user_behavior_raw ub
         |LEFT JOIN cheating_users cu ON ub.user_id = cu.user_id
         |WHERE cu.user_id IS NULL -- 排除刷单用户
         |  AND ub.action_ts >= DATE_SUB(CURRENT_DATE(), 30) -- 近30天行为{insert\_element\_6\_}
       """.stripMargin)
        }

        /**
         * 清洗体重数据：单位统一、异常值过滤、多源数据打标
         * 处理规则：斤→kg（除以2）、克→kg（除以1000），剔除<30kg或>200kg{insert\_element\_7\_}
         */
        def cleanWeightData(tEnv: StreamTableEnvironment): Table = {
        tEnv.sqlQuery(
        s"""
         |-- 合并多源体重数据（设备同步/订单评价/主动填写/活动表单）
         |WITH multi_source_weight AS (
         |  -- 1. 设备同步体重（权重1.0）
         |  SELECT
         |    user_id,
         |    sync_weight AS weight_val,
         |    'device' AS data_source,
         |    1.0 AS source_weight,
         |    sync_time AS data_time
         |  FROM dwd_device_data_raw
         |  WHERE sync_weight IS NOT NULL
         |  UNION ALL
         |  -- 2. 订单评价体重（权重0.8）
         |  SELECT
         |    user_id,
         |    comment_weight AS weight_val,
         |    'order' AS data_source,
         |    0.8 AS source_weight,
         |    pay_time AS data_time
         |  FROM dwd_order_info_raw
         |  WHERE comment_weight IS NOT NULL
         |  UNION ALL
         |  -- 3. 主动填写体重（权重0.6）
         |  SELECT
         |    user_id,
         |    weight_self AS weight_val,
         |    'profile' AS data_source,
         |    0.6 AS source_weight,
         |    register_time AS data_time
         |  FROM dwd_user_profile_raw
         |  WHERE weight_self IS NOT NULL
         |  UNION ALL
         |  -- 4. 活动表单体重（权重0.4）
         |  SELECT
         |    user_id,
         |    activity_weight AS weight_val,
         |    'activity' AS data_source,
         |    0.4 AS source_weight,
         |    submit_time AS data_time
         |  FROM ods_taobao_activity_form
         |  WHERE activity_weight IS NOT NULL
         |),
         |-- 单位统一与异常值过滤
         |standardized_weight AS (
         |  SELECT
         |    user_id,
         |    CASE
         |      WHEN weight_val LIKE '%斤%' THEN CAST(REPLACE(weight_val, '斤', '') AS DOUBLE) / 2
         |      WHEN weight_val LIKE '%g%' OR weight_val LIKE '%克%' THEN CAST(REPLACE(REPLACE(weight_val, 'g', ''), '克', '') AS DOUBLE) / 1000
         |      ELSE CAST(weight_val AS DOUBLE) -- 无单位默认kg{insert\_element\_8\_}
         |    END AS weight_kg,
         |    data_source,
         |    source_weight,
         |    data_time
         |  FROM multi_source_weight
         |  -- 剔除异常值：<30kg或>200kg{insert\_element\_9\_}
         |  WHERE CASE
         |          WHEN weight_val LIKE '%斤%' THEN CAST(REPLACE(weight_val, '斤', '') AS DOUBLE) / 2
         |          WHEN weight_val LIKE '%g%' OR weight_val LIKE '%克%' THEN CAST(REPLACE(REPLACE(weight_val, 'g', ''), '克', '') AS DOUBLE) / 1000
         |          ELSE CAST(weight_val AS DOUBLE)
         |        END BETWEEN 30 AND 200
         |    AND weight_val REGEXP '^[0-9]+(\\.[0-9]+)?(kg|斤|g|克)?$' -- 排除非数值字符{insert\_element\_10\_}
         |)
         |-- 输出清洗后的体重明细（含数据源权重）
         |SELECT
         |  user_id,
         |  weight_kg,
         |  data_source,
         |  source_weight,
         |  data_time
         |FROM standardized_weight
       """.stripMargin)
        }

        /**
         * 清洗身高数据：异常值过滤、尺码推导补全
         * 处理规则：剔除<100cm或>250cm，服饰类目用户用尺码表推算{insert\_element\_11\_}
         */
        def cleanHeightData(tEnv: StreamTableEnvironment): Table = {
        tEnv.sqlQuery(
        s"""
         |-- 合并多源身高数据（设备同步/实名认证/订单尺码/客服记录）
         |WITH multi_source_height AS (
         |  -- 1. 设备同步身高（权重1.0）
         |  SELECT
         |    user_id,
         |    sync_height AS height_cm,
         |    'device' AS data_source,
         |    1.0 AS source_weight,
         |    sync_time AS data_time
         |  FROM dwd_device_data_raw
         |  WHERE sync_height IS NOT NULL
         |  UNION ALL
         |  -- 2. 实名认证身高（权重0.9）
         |  SELECT
         |    user_id,
         |    height_self AS height_cm,
         |    'profile' AS data_source,
         |    0.9 AS source_weight,
         |    register_time AS data_time
         |  FROM dwd_user_profile_raw
         |  WHERE height_self IS NOT NULL
         |  UNION ALL
         |  -- 3. 订单尺码推导身高（权重0.85，近30天）
         |  SELECT
         |    user_id,
         |    CASE
         |      WHEN gender = 'male' THEN -- 男装尺码：L=175, XL=180{insert\_element\_12\_}
         |        CASE size WHEN 'L' THEN 175 WHEN 'XL' THEN 180 WHEN 'M' THEN 170 ELSE NULL END
         |      WHEN gender = 'female' THEN -- 女装尺码：L=170, XL=175{insert\_element\_13\_}
         |        CASE size WHEN 'L' THEN 170 WHEN 'XL' THEN 175 WHEN 'M' THEN 165 ELSE NULL END
         |      ELSE NULL
         |    END AS height_cm,
         |    'order_size' AS data_source,
         |    0.85 AS source_weight,
         |    pay_time AS data_time
         |  FROM dwd_order_info_raw oi
         |  JOIN dim_user_gender ug ON oi.user_id = ug.user_id -- 关联性别标签（提前计算）
         |  WHERE oi.category_id IN (SELECT category_id FROM dim_clothing_category) -- 服饰类目
         |    AND oi.pay_time >= DATE_SUB(CURRENT_DATE(), 30)
         |)
         |-- 异常值过滤：剔除<100cm、>250cm、非整数{insert\_element\_14\_}
         |SELECT
         |  user_id,
         |  CAST(height_cm AS INT) AS height_cm, -- 过滤非整数
         |  data_source,
         |  source_weight,
         |  data_time
         |FROM multi_source_height
         |WHERE height_cm BETWEEN 100 AND 250
         |  AND height_cm = CAST(height_cm AS INT)
       """.stripMargin)
        }

        // 注册DWD层清洗后的表
        def registerDwdTables(tEnv: StreamTableEnvironment): Unit = {
        tEnv.createTemporaryView("dwd_clean_user_behavior", cleanUserBehavior(tEnv))
        tEnv.createTemporaryView("dwd_clean_weight", cleanWeightData(tEnv))
        tEnv.createTemporaryView("dwd_clean_height", cleanHeightData(tEnv))
        // 注册星座计算所需的生日清洗表（提取月份日期）
        tEnv.createTemporaryView("dwd_clean_birthday",
        tEnv.sqlQuery("""SELECT user_id,
                          DATE_FORMAT(TO_TIMESTAMP(birthday, 'yyyy-MM-dd'), 'MM-dd') AS mmdd,
                          birthday AS raw_birthday
                       FROM dwd_user_profile_raw
                       WHERE birthday IS NOT NULL"""))
        }
        }